package cn.creditease.std.producer.impl;

import cn.creditease.std.consumer.JobConsumer;
import cn.creditease.std.consumer.JobConsumerRegistory;
import cn.creditease.std.context.JobAlloterContext;
import cn.creditease.std.model.Job;
import cn.creditease.std.producer.Producer;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;
import java.io.IOException;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.util.CollectionUtils;

/**
 * @Description: 通过mq队列做消费通道，此出主要起到通知到作用，采用广播模式，发布消息
 * @Author: kai.yang
 * @Date: 2019-08-07 10:49
 */
public class MqProducerImpl implements Producer {


    protected ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);


    private static final Logger logger = LoggerFactory.getLogger(MqProducerImpl.class);

    final RabbitTemplate rabbitTemplate;

    final ConnectionFactory connectionFactory;

    com.rabbitmq.client.Channel channel;


    /**
     * mq交互机命名
     */
    private static final String ALLOT_MQ_CHANNEL_EXCHANGE = "allot_mq_channel_exchange";


    /**
     * mq消息队列命名
     */
    private static final String ALLOT_MQ_CHANNEL_ROUNTING = "allot_mq_channel_rounting";

    /**
     * 默认队列命名
     */
    private static final String ALLOT_MQ_CHANNEL_QUEUE_PRE = "allot_mq_channel_queue_";

    /**
     * mq采用广播模式进行通知各消费端，各消费端可以定义自己的tag，此处每个tag创建单独队列来接收消息通知
     */
    private String getLocalQueueName() {
        return ALLOT_MQ_CHANNEL_QUEUE_PRE + JobAlloterContext.getContext().getCustomerId();
    }


    public MqProducerImpl(RabbitTemplate rabbitTemplate,
        ConnectionFactory connectionFactory) throws IOException {
        this.rabbitTemplate = rabbitTemplate;
        this.connectionFactory = connectionFactory;
        Connection connection = connectionFactory.createConnection();

        channel = connection.createChannel(true);

        /**
         * 开启队列监听消费
         */
        openLisenter();

        /**
         * 开启mq通道断开后的监听处理，但connectionFactory工厂重连，又业务服务自己实现
         */
        scheduledThreadPoolExecutor.scheduleAtFixedRate(() -> recoverMqChannle(), 60, 60 * 30, TimeUnit.SECONDS);
    }


    /**
     * 开启队列监听
     */
    private void openLisenter() throws IOException {

        channel.exchangeDeclare(ALLOT_MQ_CHANNEL_EXCHANGE, BuiltinExchangeType.FANOUT, true, true, null);

        logger.info("创建分发消费者{}", getLocalQueueName());

        channel.exchangeDeclare(ALLOT_MQ_CHANNEL_EXCHANGE, BuiltinExchangeType.FANOUT, true, true, null);

        logger.info("创建分发消费者{}", getLocalQueueName());
        // 声明队列
        channel.queueDeclare(getLocalQueueName(), true, false, true, null);

        // 绑定队列和交换机
        channel.queueBind(getLocalQueueName(), ALLOT_MQ_CHANNEL_EXCHANGE, ALLOT_MQ_CHANNEL_ROUNTING);

        //消费者，开始接收消息,配置响应为自动应答
        channel.basicConsume(getLocalQueueName(), true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                byte[] body) {
                try {
                    String jobStr = new String(body, "UTF-8");
                    logger.info("消费分发任务{}", jobStr);

                    String jobId;
                    String jobName;
                    String[] split = jobStr.split("&");
                    if (split.length != 2) {
                        return;
                    } else {
                        jobId = split[0];
                        jobName = split[1];
                    }
                    JobConsumer consumer = JobConsumerRegistory.getConsumer(jobName);
                    if (consumer != null) {
                        consumer.consume(jobId, jobName);
                    }


                } catch (Throwable e) {
                    logger.error("分发消费异常", e);
                    //do nothing
                } finally {
                }
            }

        });
    }

    /**
     * 通道断线重连机制
     */
    private void recoverMqChannle() {
        try {
            if (channel == null || !channel.isOpen()) {
                this.channel = connectionFactory.createConnection().createChannel(true);
                openLisenter();
            }
        } catch (Exception e) {
            logger.error("mq消费端通道断开，重连失败", e);
        }
    }

    @Override
    public void send(List<Job> jobList) {
        Set<String> names = new HashSet<>();
        if (!CollectionUtils.isEmpty(jobList)) {
            for (Job job : jobList) {
                if (!names.contains(job.getJobId())) {
                    names.add(job.getJobId());
                    rabbitTemplate
                        .convertAndSend(ALLOT_MQ_CHANNEL_EXCHANGE, ALLOT_MQ_CHANNEL_ROUNTING,
                            job.getJobId() + "&" + job.getJobGroupName());
                }
            }
        }
        names.clear();
    }


    @Override
    public String getType() {
        return "mq";
    }


    @Override
    public int getOrder() {
        return 1;
    }
}
